Stock Analysis MCP Server

# FastMCP + CherryStudio：手把手教你打造专属AI"股票分析"助手！ ## 🚀 前言 在AI时代，如何让大模型具备实时获取股票数据的能力？今天我们将使用**FastMCP**框架和**CherryStudio**客户端，手把手教你打造一个专属的AI股票分析助手！ 通过本教程，你将学会： - 使用FastMCP框架构建MCP服务器 - 集成AKShare获取实时股票数据 - 在CherryStudio中配置和使用自定义工具 - 实现SSE协议的实时数据流 ## 🎯 项目效果展示 我们的股票分析助手可以： - 📈 **实时股票行情查询**：输入股票代码，获取最新价格、涨跌幅等信息 - 📊 **市场情绪分析**：分析整体市场的涨跌情况和投资情绪 - 🔄 **实时数据流**：基于SSE协议，支持实时数据推送 ## 🛠️ 技术架构 ### 核心技术栈 - **FastMCP**: 官方推荐的MCP服务器框架 - **Starlette**: 高性能异步Web框架 - **AKShare**: 开源金融数据接口 - **SSE协议**: 服务器发送事件，实现实时通信 ### 架构图 ``` CherryStudio客户端 ←→ SSE协议 ←→ FastMCP服务器 ←→ AKShare API ``` ## 📝 核心代码实现 ### 1. 服务器主体代码 ```python #!/usr/bin/env python3 """ 股票分析SSE服务器 - 官方推荐方案最终版 基于Starlette和mcp.py库内置的SseServerTransport，确保协议兼容性。 """ import json import logging import akshare as ak import uvicorn from datetime import datetime from mcp.server.fastmcp import FastMCP from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Mount, Route # --- 配置 --- logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- MCP服务器初始化 --- mcp = FastMCP("StockAnalysisServer") ``` ### 2. 股票实时数据工具 ```python @mcp.tool() async def get_stock_realtime_data(symbol: str) -> str: """ 获取指定股票代码的实时行情数据。 :param symbol: 股票代码, 例如 '000001' :return: JSON格式的股票数据字符串 """ logger.info(f"执行工具 get_stock_realtime_data，参数: {symbol}") try: df = ak.stock_zh_a_spot_em() stock_data = df[df['代码'] == symbol] if stock_data.empty: result = {"error": f"未找到股票代码 {symbol} 的数据"} else: data = stock_data.iloc[0] result = { "股票代码": symbol, "股票名称": data['名称'], "最新价": float(data['最新价']), "涨跌幅": float(data['涨跌幅']), "更新时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } return json.dumps(result, ensure_ascii=False) except Exception as e: logger.error(f"获取实时数据失败: {e}") error_result = {"error": f"获取实时数据失败: {str(e)}"} return json.dumps(error_result, ensure_ascii=False) ``` ### 3. 市场情绪分析工具 ```python @mcp.tool() async def get_market_sentiment() -> str: """ 获取当前市场的整体情绪分析。 :return: JSON格式的市场情绪数据字符串 """ logger.info("执行工具 get_market_sentiment") try: df_market = ak.stock_zh_a_spot_em() up_count = len(df_market[df_market['涨跌幅'] > 0]) down_count = len(df_market[df_market['涨跌幅'] < 0]) total_count = len(df_market) sentiment = "乐观" if up_count > down_count else "悲观" if down_count > up_count else "中性" result = { "上涨股票数": up_count, "下跌股票数": down_count, "市场情绪": sentiment, "分析时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } return json.dumps(result, ensure_ascii=False) except Exception as e: logger.error(f"获取市场情绪失败: {e}") error_result = {"error": f"获取市场情绪失败: {str(e)}"} return json.dumps(error_result, ensure_ascii=False) ``` ### 4. SSE服务器配置 ```python # --- SSE传输和Starlette应用设置 --- sse_transport = SseServerTransport('/messages/') async def sse_handler(request): """处理初始SSE连接并运行MCP服务器会话。""" async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams: await mcp._mcp_server.run(streams[0], streams[1], mcp._mcp_server.create_initialization_options()) # 定义Starlette应用的路由 app = Starlette( debug=True, routes=[ Route('/sse', endpoint=sse_handler), Mount('/messages/', app=sse_transport.handle_post_message) ] ) ``` ### 5. 服务器启动代码 ```python if __name__ == "__main__": port = 8001 # 打印FastMCP logo print(""" ███████╗ █████╗ ███████╗████████╗███╗ ███╗ ██████╗██████╗ ██╔════╝██╔══██╗██╔════╝╚══██╔══╝████╗ ████║██╔════╝██╔══██╗ █████╗ ███████║███████╗ ██║ ██╔████╔██║██║ ██████╔╝ ██╔══╝ ██╔══██║╚════██║ ██║ ██║╚██╔╝██║██║ ██╔═══╝ ██║ ██║ ██║███████║ ██║ ██║ ╚═╝ ██║╚██████╗██║ ╚═╝ ╚═╝ ╚═╝╚══════╝ ╚═╝ ╚═╝ ╚═╝ ╚═════╝╚═╝ """) logger.info("启动基于官方推荐方案的股票分析SSE服务器...") logger.info(f"服务器名称: 股票分析MCP服务器") logger.info(f"服务器版本: v1.0.0") logger.info(f"描述: 基于AKShare API的股票分析工具，支持SSE协议") logger.info(f"MCP主端点: http://127.0.0.1:{port}/sse") uvicorn.run(app, host="127.0.0.1", port=port) ``` ## 🔧 环境配置 ### 1. 安装依赖 ```bash pip install fastmcp akshare pandas starlette uvicorn ``` ### 2. 项目结构 ``` stock_mcp/ ├── server.py # 主服务器文件 ├── requirements.txt # 依赖包列表 └── README.md # 项目说明 ``` ### 3. requirements.txt ```txt fastmcp>=0.2.0 akshare>=1.12.0 pandas>=2.0.0 starlette>=0.27.0 uvicorn>=0.23.1 ``` ## 🎮 CherryStudio配置 ### 1. 启动服务器 ```bash python server.py ``` 看到以下输出说明启动成功： ``` ███████╗ █████╗ ███████╗████████╗███╗ ███╗ ██████╗██████╗ INFO: 启动基于官方推荐方案的股票分析SSE服务器... INFO: MCP主端点: http://127.0.0.1:8001/sse INFO: Uvicorn running on http://127.0.0.1:8001 ``` ### 2. CherryStudio配置 在CherryStudio中添加MCP服务器： - **名称**: 股票分析SSE工具 - **类型**: 服务器发送事件 (sse) - **URL**: `http://127.0.0.1:8001/sse` - **超时**: 120秒 ### 3. 使用示例 配置完成后，你可以在CherryStudio中这样使用： **查询股票实时数据：** ``` 请帮我查询平安银行(000001)的实时股价 ``` **分析市场情绪：** ``` 请分析一下当前A股市场的整体情绪 ``` ## 📊 实际运行效果 从服务器日志可以看到，系统已经成功处理了多个请求： ``` INFO: Processing request of type CallToolRequest INFO: 执行工具 get_stock_realtime_data，参数: 000001 INFO: 执行工具 get_stock_realtime_data，参数: 600519 ``` 这说明我们的股票分析助手已经在正常工作！ ## 🎯 核心技术要点 ### 1. MCP协议优势 - **标准化**: 遵循官方MCP协议规范 - **可扩展**: 易于添加新的分析工具 - **实时性**: 基于SSE协议的实时数据流 ### 2. FastMCP框架特点 - **简单易用**: 装饰器语法，开发效率高 - **官方支持**: mcp.py库官方推荐方案 - **协议兼容**: 完美支持各种MCP客户端 ### 3. 数据源选择 - **AKShare**: 免费开源的金融数据接口 - **实时性**: 支持A股实时行情数据 - **丰富性**: 涵盖股票、基金、期货等多种数据 ## 🚀 扩展功能 基于当前架构，你还可以轻松扩展： - 📈 **技术指标分析**: 添加MA、MACD、RSI等技术指标 - 📰 **新闻情感分析**: 集成新闻API，分析市场情绪 - 🔔 **价格预警**: 实现股价异动提醒功能 - 📱 **多市场支持**: 扩展到港股、美股等市场 ## 💡 总结 通过本教程，我们成功构建了一个功能完整的AI股票分析助手： 1. **技术先进**: 使用FastMCP + SSE协议，确保实时性和稳定性 2. **功能实用**: 实时行情查询和市场情绪分析，满足基本投资需求 3. **易于扩展**: 模块化设计，方便添加新功能 4. **开源免费**: 基于开源技术栈，成本低廉 希望这个项目能帮助你更好地理解MCP协议的应用，也为你的AI工具开发提供参考！ --- **项目地址**: [GitHub链接] **技术交流**: 欢迎在评论区讨论技术问题 **关注我们**: 获取更多AI开发教程 #FastMCP #CherryStudio #AI工具 #股票分析 #Python开发